GH-48476: [C++] [Parquet] Add max_row_group_size writer property to limit row groups by compressed byte size - #50743
GH-48476: [C++] [Parquet] Add max_row_group_size writer property to limit row groups by compressed byte size#50743zhf999 wants to merge 4 commits into
Conversation
|
Thanks for opening a pull request! This pull request has been automatically converted to a draft because its title doesn't match Arrow's required format. If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project. Then could you also rename the pull request title in the following format? or After updating the title, you can mark the pull request as ready for review. See also: |
zhf999
changed the title
|
|
|
Related PR: #48468 |
Thanks for contributing! Can you edit the PR desc and ref to the issue, it seems a PR is being referenced here. |
| /// The limit is checked against the compressed pages accumulated in the | ||
| /// current row group, so the actual row group size may slightly exceed it. | ||
| /// Only effective for buffered row groups ( | ||
| /// parquet::arrow::FileWriter::WriteRecordBatch). |
There was a problem hiding this comment.
WriteRecordBatch writes into a buffered row group, which accumulates across calls. That makes the byte size observable between writes, so we can simply stop adding to the current row group once the accumulated compressed size reaches the limit.
While WriteTable uses the non-buffered path, where each call maps to exactly one row group whose row count is fixed up front by the user-supplied chunk_size. Enforcing a byte limit there would require predicting the compressed size before writing, e.g. by estimating an average row size from the previous row group. That's inherently approximate, especially since chunk_size is an explicit contract from the caller.
I wonder if estimation is acceptable in WriteTable?
There was a problem hiding this comment.
IIRC, WriteTable also splits the table into several row groups if max_row_group_length is reached.
There was a problem hiding this comment.
max_row_group_length can be applied there because row counts are known before writing — it just clamps chunk_size. The byte size simply isn't knowable at that point, which is the asymmetry.
There was a problem hiding this comment.
@wgtmac What about change the WriteTable to buffered path like WriteRecordBatch?
|
|
github-actions
Bot
added
awaiting committer review
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (3)
cpp/src/parquet/arrow/writer.cc:439
- This changes FileWriter::WriteTable behavior when max_row_group_size is set (it switches to the buffered RecordBatch path and will split row groups by byte size). The PR description currently states WriteTable is intentionally not affected; please update the PR description (and any related docs/release notes) to match the implemented behavior to avoid misleading users.
// If max_row_group_size is set, use buffered path to write row groups.
if (this->properties().max_row_group_size() != std::numeric_limits<int64_t>::max()) {
::arrow::TableBatchReader reader(table);
reader.set_chunksize(std::min(chunk_size, this->properties().write_batch_size()));
while (true) {
cpp/src/parquet/arrow/writer.cc:496
- estimated_row_group_size() calls total_compressed_bytes(), total_compressed_bytes_written(), and estimated_buffered_stats(); each of these does an O(num_columns) scan in RowGroupSerializer. With max_row_group_size enabled, row_group_full() is evaluated after each WriteBatch, so this becomes 2–3 full column scans per batch and can be a noticeable CPU cost for wide schemas.
// Estimated size of the data accumulated in the current row group.
auto estimated_row_group_size = [&]() {
const auto buffered = row_group_writer_->estimated_buffered_stats();
return row_group_writer_->total_compressed_bytes() +
row_group_writer_->total_compressed_bytes_written() + buffered.value_bytes +
buffered.def_level_bytes + buffered.rep_level_bytes + buffered.dict_bytes;
};
cpp/src/parquet/properties.h:506
- The doc comment claims the size estimate “errs on the conservative side”, but the implementation doesn’t account for per-page overhead (e.g., PageHeader bytes) for data still buffered in encoders, so small pages / low compression can lead to underestimation. Consider softening this guarantee to avoid overpromising the bound.
/// The limit is checked between writes against an estimate of the data
/// accumulated in the current row group, which combines the size of the
/// compressed pages with an uncompressed estimate of the values still
/// buffered by the column encoders. The estimate is therefore approximate
/// and errs on the conservative side.
|
|
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (2)
cpp/src/parquet/arrow/writer.cc:503
row_group_full()always computesestimated_buffered_stats()and callstotal_compressed_bytes*()even whenmax_row_group_sizeis unlimited. SinceWriteRecordBatch()now always goes throughWriteRecordBatchBuffered(), this adds an O(num_columns) scan per chunk on the default (unlimited) path and can be a noticeable regression.
auto row_group_full = [&]() {
const auto buffered = row_group_writer_->estimated_buffered_stats();
const int64_t estimated_size = row_group_writer_->total_compressed_bytes() +
row_group_writer_->total_compressed_bytes_written() +
buffered.value_bytes + buffered.def_level_bytes +
cpp/src/parquet/arrow/writer.cc:545
- When
max_row_group_sizeis set, size enforcement is only checked after eachWriteBatch()call, butbatch_sizecan be as large as the remaining rows up tomax_rows_per_row_group(often 1M). A single largeRecordBatchcan therefore overshoot the byte limit by a very large margin before the rollover check runs.
const int64_t batch_size =
std::min(max_rows_per_row_group - row_group_writer_->num_rows(),
batch.num_rows() - offset);
RETURN_NOT_OK(WriteBatch(offset, batch_size));
Rationale for this change
The Parquet writer can currently only limit row groups by row count
(
WriterProperties::max_row_group_length()). With wide rows or highly variablerow sizes, a row-count limit produces row groups whose size in bytes varies a
lot. Since query engines and storage systems usually tune around a target row
group size in bytes (HDFS block alignment, reader memory footprint,
parallelism granularity), it is useful to be able to limit row groups by size
as well.
What changes are included in this PR?
New writer property
WriterProperties::max_row_group_size()/Builder::max_row_group_size(int64_t)specify the maximum size of a row group in bytes. The default is unlimited
(
std::numeric_limits<int64_t>::max()), so nothing changes for existing usersunless they opt in.
How the size is estimated
The limit is checked between writes against the data accumulated in the current
row group:
(
RowGroupWriter::total_compressed_bytes_written()),(
RowGroupWriter::total_compressed_bytes()),and dictionary still held by the column encoders
(
RowGroupWriter::estimated_buffered_stats()).The last group matters: values only become pages once they exceed
data_pagesize, so ignoring the encoder buffers would let a row groupovershoot the limit by up to
num_columns * data_pagesizeand would make anylimit below that value impossible to honour. Those estimates are uncompressed,
so the total errs on the conservative side and row groups stay at or below the
limit in practice.
Enforcement in both Arrow writer entry points
WriteRecordBatchstarts a new buffered row group whenever the current onereaches the row count or the byte size limit. The implementation moved into a
private
WriteRecordBatchBuffered(batch, max_rows_per_row_group)so that therow cap can be supplied by the caller.
WriteTableswitches to the buffered path when the limit is set: the table isfed through a
TableBatchReaderin batches ofmin(chunk_size, write_batch_size())rows, so the same check applies and arow group only overshoots by at most one batch.
chunk_sizeis still honouredas the maximum number of rows per row group. Without the limit,
WriteTablekeeps using the original non-buffered path unchanged.
A byte limit cannot be enforced on non-buffered row groups because their size
is only known once they have been written, and the row count is fixed before
writing. Row groups created explicitly through
NewRowGroup()/WriteColumnChunk()are therefore not affected; this isdocumented on the builder method.
Are these changes tested?
Yes.
WriterPropertiesTest.RoundTripThroughBuildercovers the new property, with anon-default value added to the
override_defaultscase.TestArrowReadWrite.WriteRecordBatchRespectsMaxRowGroupSizeandTestArrowReadWrite.WriteTableRespectsMaxRowGroupSizecheck that row groupsare rolled over once the limit is reached, that every row group stays within
the limit, and that no row is lost or duplicated. Both use the default 1MB
data page size, so they also cover the encoder-buffered data.
TestArrowReadWrite.WriteTableUnlimitedRowGroupSizepins the defaultbehaviour, where
chunk_sizealone decides the row group boundaries.TestArrowReadWrite.WriteTableMaxRowGroupSizeRoundTripverifies that the datais unchanged when written through the buffered path.
Are there any user-facing changes?
Yes, a new opt-in writer property. The default is unlimited, so there is no
behaviour change unless it is set. When it is set:
memory footprint of the writer;
WriteTable, row group boundaries are decided by bothchunk_sizeandthe byte limit, and columns are written in parallel if
ArrowWriterProperties::use_threads()is enabled.